import org.apache.spark.SparkConf
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.types.IntegerType

import java.util.Properties

object Test {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("movie_test")
      .setMaster("local[*]")

    val spark = SparkSession
      .builder()
      .config(conf)
      .enableHiveSupport()
      .getOrCreate()

    val mysqlConf = new Properties()
    val url = "jdbc:mysql://single01:3306/ebs?useUnicode=true&charSet=utf-8"
    mysqlConf.setProperty("url", url)
    mysqlConf.setProperty("user", "root")
    mysqlConf.setProperty("password", "L3154638821")
    mysqlConf.setProperty("driver", "com.mysql.cj.jdbc.Driver")

    import spark.implicits._
    import org.apache.spark.sql.functions._

    /**
     *  电影总体指标：电影数量, 平均评分, 平均评论人数, 平均时长
     */
    spark
      .table("movies.movie")
      .agg(
        count("*").as("movie_cnt"),
        avg($"rating").as("avg_rating"),
        avg("rating_count").as("avg_rat_cnt"),
        avg(regexp_extract($"runtime"(0), "(\\d+)", 1).cast(IntegerType)).as("avg_runtime")
      )
      .write
      .mode(SaveMode.Overwrite)
      .jdbc(url, "movie_main", mysqlConf)

  }
}
